package org.corf.endpoint.collector;

import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;
import org.corf.endpoint.JmsPayloadHeaderKey;
import org.corf.payload.Payload;
import org.corf.payload.PayloadType;
import org.corf.processor.CorfProcessor;

/**
 * Abstract JMS acceptor implementation that provides the basic infrastructure.
 */
public class JmsCollector extends AbstractCollector implements MessageListener {

    // -------------------------------------------------------------------------------------------------------------------------------------
    // Class constants.
    // -------------------------------------------------------------------------------------------------------------------------------------

    public static final Logger LOG = Logger.getLogger(JmsCollector.class);

    // -------------------------------------------------------------------------------------------------------------------------------------
    // Class methods.
    // -------------------------------------------------------------------------------------------------------------------------------------

    protected static Payload createPayloadFromMessage(Message message) throws JMSException {
        String payloadType = message.getStringProperty(JmsPayloadHeaderKey.PAYLOAD_TYPE.getMessagePropertyKey());
        if (payloadType == null || payloadType.length() == 0) {
            throw new IllegalArgumentException("Missing message property '" + JmsPayloadHeaderKey.PAYLOAD_TYPE.getMessagePropertyKey()
                    + "'!");
        }
        String payloadId = message.getStringProperty(JmsPayloadHeaderKey.PAYLOAD_ID.getMessagePropertyKey());
        if (payloadId == null || payloadId.length() == 0) {
            throw new IllegalArgumentException("Missing message property '" + JmsPayloadHeaderKey.PAYLOAD_ID.getMessagePropertyKey() + "'!");
        }
        PayloadType pt = PayloadType.valueOf(payloadType);
        Payload p = Payload.create(pt, payloadId);

        return p;
    }

    @SuppressWarnings("unchecked")
    protected static void extractMessageProperties(Message m, Payload p) throws JMSException {
        // Extract generic properties.
        Enumeration<String> e = m.getPropertyNames();
        if (e == null || !e.hasMoreElements()) {
            return;
        }
        String value = null;
        for (String key = e.nextElement(); e.hasMoreElements();) {
            try {
                value = m.getStringProperty(key);
                p.setHeader(key, value);
            } catch (Exception ex) {
                // In case that a conversion of a property value into a string value failed, we just skip it, since we only support string
                // values in our payload header map.
            }
        }
        // Extract well defined properties.
        // TODO the implementation.
    }

    public static Payload extractPayloadFromMessage(Message message) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Extracting payload from Message.");
        }
        Payload result = null;
        return result;
    }

    public static Payload extractPayloadFromBytesMessage(BytesMessage message) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Extracting payload from BytesMessage.");
        }
        Payload result = null;
        return result;
    }

    public static Payload extractPayloadFromMapMessage(MapMessage message) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Extracting payload from MapMessage.");
        }
        Payload result = null;
        return result;
    }

    public static Payload extractPayloadFromObjectMessage(ObjectMessage message) throws JMSException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Extracting payload from ObjectMessage.");
        }
        Payload result = null;
        try {
            final Object o = message.getObject();
            if (o instanceof Payload) {
                result = (Payload) o;
            }
        } catch (final JMSException ex) {
            LOG.warn("Something went wrong on extracting the payload: " + ex.getMessage(), ex);
            throw ex;
        }
        return result;
    }

    public static Payload extractPayloadFromStreamMessage(StreamMessage message) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Extracting payload from StreamMessage.");
        }
        Payload result = null;
        return result;
    }

    public static Payload extractPayloadFromTextMessage(TextMessage message) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Extracting payload from TextMessage.");
        }
        Payload result = null;
        return result;
    }

    public static Payload extractPayload(Message message) {
        Payload result = null;
        try {
            if (message instanceof TextMessage) {
                result = extractPayloadFromTextMessage((TextMessage) message);
            } else if (message instanceof StreamMessage) {
                result = extractPayloadFromStreamMessage((StreamMessage) message);
            } else if (message instanceof ObjectMessage) {
                result = extractPayloadFromObjectMessage((ObjectMessage) message);
            } else if (message instanceof ObjectMessage) {
                result = extractPayloadFromMapMessage((MapMessage) message);
            } else if (message instanceof ObjectMessage) {
                result = extractPayloadFromBytesMessage((BytesMessage) message);
            } else if (message instanceof ObjectMessage) {
                result = extractPayloadFromMessage(message);
            }
        } catch (Exception ex) {
            LOG.error("Unfortunately, payload extraction failed: " + ex.getMessage());
        }
        return result;
    }

    protected final List<CorfProcessor> processors;

    protected JmsCollector(String name) {
        super(name);
        this.processors = new ArrayList<CorfProcessor>();
    }

    protected JmsCollector(String name, List<CorfProcessor> processors) {
        super(name);
        this.processors = processors;
    }

    public void addProcessor(CorfProcessor processor) {
        processors.add(processor);
    }

    public String listProcessors() {
        return processors.toString();
    }

    @Override
    public void onPayload(Payload p) {
        // TODO Auto-generated method stub
    }

    @Override
    public void onMessage(Message message) {
        Payload p = extractPayload(message);
        signPayload(p);
    }

}
